package top.wangjiu.reactive.rpc.handler;

import org.reactivestreams.Subscription;
import reactor.core.publisher.Mono;
import top.wangjiu.reactive.rpc.model.RpcRequest;
import top.wangjiu.reactive.rpc.model.RpcResponse;

/**
 * @ClassName MonoResultTypeHandler
 * @Description TODO
 * @Author wangjiu
 * @Date 2021/8/3 5:03 下午
 */
public class MonoRequestHandler extends AbstractRequestHandler<Mono> {

    private volatile Mono res;

    private volatile String requestId;

    private volatile Subscription subscription;

    @Override
    public void onRequest(RpcRequest request) {
        if(requestId == null){
            requestId = request.getRequestId();
            RpcResponse response = new RpcResponse(requestId);
            response.setFirst(true);
            response.setLast(true);
            try {
                res = invoke(request);
                res.subscribe(r->{
                   response.setResult(r);
                   write(response);
                },throwable -> {
                    response.setSuccess(false);
                    response.setCause((Throwable) throwable);
                    write(response);
                },()->{
                    write(response);
                },s -> {
                    subscription = (Subscription) s;
                    if(request.isCancel()){
                        subscription.cancel();
                    }else {
                        subscription.request(request.getCount());
                    }
                });
            } catch (Exception e) {
                response.setSuccess(false);
                response.setCause(e);
                write(response);
            }
        }else{
            if(subscription !=null){
                if(request.isCancel()){
                    subscription.cancel();
                }else {
                    subscription.request(request.getCount());
                }
            }
        }
    }

    @Override
    public void onDisconnect() {
        if(res!=null){

        }
    }
}
